/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.
 */

import cluster from 'cluster';
import EventEmitter from 'events';
import { Socket } from 'socket.io';

import { BuildEvent, WatchEvent } from '../../common/daemon-protocol/build-event.js';
import { OutputType } from '../../common/daemon-protocol/common-enum.js';
import { createOutput } from '../../common/daemon-protocol/message.js';
import { HvigorLogger } from '../../log/hvigor-log.js';

import { getThreadId, resolveThreadId, SessionMap, SocketId, ThreadId } from './session-map.js';

const _log = HvigorLogger.getLogger('daemon');
export const SESSION_EMIT_LOG = 'emitLog';

class SessionManager extends EventEmitter {
  private sessionMap: SessionMap;
  private readonly serverSockets: Socket[];

  // 指示当前活跃于主线程的socket，主线程将向curActiveSocket发送消息。
  private curActiveSocket: Socket | undefined;

  constructor() {
    super();
    this.sessionMap = new SessionMap();
    this.serverSockets = [];
    this.addListenersOnSocket();
  }

  /**
   * 添加socket与相对应的session，添加session条目与socket记录
   * @param {Socket} socket
   */
  initSession(socket: Socket): void {
    this.serverSockets.push(socket);

    // 添加session条目
    this.sessionMap.setSession(socket.id, undefined);
  }

  /**
   * 结束socket所对应的session，移除session条目和socket记录
   * @param {Socket} socket
   */
  endSession(socket: Socket): void {
    const index = this.serverSockets.indexOf(socket);
    if (index !== -1) {
      // 从socketArray中移除socket
      this.serverSockets.splice(index, 1);

      // 移除session条目
      this.sessionMap.deleteBySocketId(socket.id);
    }
  }

  /**
   * 绑定socketId与threadId
   *
   * @param {SocketId} socketId
   * @param {ThreadId} threadId
   */
  bindThread(socketId: SocketId, threadId: ThreadId) {
    if (this.sessionMap.checkBySocketId(socketId)) {
      this.sessionMap.setSession(socketId, threadId);
    }
  }

  /**
   * 绑定socketId与worker process
   *
   * @param {SocketId} socketId
   * @param workerId
   */
  bindProcess(socketId: SocketId, workerId: number) {
    if (this.sessionMap.checkBySocketId(socketId)) {
      this.sessionMap.setSession(socketId, getThreadId(workerId, 0));
    }
  }

  getSockets(): Socket[] {
    return this.serverSockets;
  }

  getLength(): number {
    return this.serverSockets.length;
  }

  getSocketByThreadId(threadId: ThreadId) {
    const socketId = this.sessionMap.getSocketId(threadId);
    return this.serverSockets.find((socket) => socket.id === socketId);
  }

  getSocketBySocketId(socketId: SocketId) {
    return this.serverSockets.find((socket) => socket.id === socketId);
  }

  hasSocket(socketId: SocketId) {
    return this.serverSockets.some((socket) => socket.id === socketId);
  }

  getThreadId(socketId: SocketId) {
    return this.sessionMap.getThreadId(socketId);
  }

  /**
   * socket收到信息时触发，给对应worker进程发送消息，目标是传递给对应worker进程内的worker线程
   *
   * @param {SocketId} socketId 触发需要传递信息的socket
   * @param {object} data 需要传递的信息
   */
  postMessageToThread(socketId: SocketId, data: object) {
    const threadId = this.getThreadId(socketId);
    if (!threadId) {
      _log.warn(`session manager: cannot find corresponding thread for socket id ${socketId}.`);
      return;
    }
    const [processId, workerThreadId] = resolveThreadId(threadId);
    if (cluster.isMaster) {
      const workerProcess = cluster.workers![processId];
      if (workerProcess === undefined) {
        _log.warn(`session manager: cannot find corresponding worker process for process id ${processId}.`);
        return;
      }
      _log.debug('session manager: send message to worker process.');
      workerProcess.send({
        ...data,
        workerId: workerThreadId,
      });
    } else {
      _log.warn('session manager: not master process, should be an error.');
    }
  }

  /**
   * 设定activeSocket，并重载输出流和错误流的write方法
   * @param {Socket} socket
   */
  setActiveSocket(socket: Socket) {
    _log.debug(`session manager: set active socket. socketId=${socket.id}`);
    this.curActiveSocket = socket;
  }

  /**
   * 重置activeSocket，并还原输出流和错误流的write方法
   */
  resetActiveSocket() {
    this.curActiveSocket = undefined;
  }

  getActiveSocket() {
    return this.curActiveSocket;
  }

  /**
   * 添加从worker process里传递过来的信息的监听器，相应地回传给socket或者其他操作
   */
  addListenersOnSocket() {
    if (cluster.isMaster) {
      cluster.on('message', (worker, response) => {
        if (!response.type) {
          return;
        }
        let socket: Socket | undefined;
        switch (response.type as string) {
          case WatchEvent.NEW_WATCH_WORKER:
            if (this.curActiveSocket) {
              const socketId = this.curActiveSocket.id;
              const threadId = getThreadId(worker.id, response.workerId);
              this.bindThread(socketId, threadId);
              _log.debug(`session manager: binding session. socketId=${socketId}, threadId=${threadId}.`);

              // 在socket上初始化注册收到需要发送给worker线程的消息的事件监听
              this.curActiveSocket.on(WatchEvent.WATCH_COMPILE_DATA, async (data) => {
                _log.debug('session manager: receive watch compile data.');
                defaultSessionManager.postMessageToThread(socketId, {
                  type: WatchEvent.WATCH_COMPILE_DATA,
                  content: data,
                });
              });

              // 在socket上初始化注册关闭worker线程的watch功能的事件监听
              this.curActiveSocket.on(WatchEvent.CLOSE_WATCH, async () => {
                _log.debug('session manager: receive close watch request.');
                defaultSessionManager.postMessageToThread(socketId, {
                  type: WatchEvent.CLOSE_WATCH,
                });
              });
            }
            break;
          case WatchEvent.WATCH_COMPILE_RESULT:
          case WatchEvent.WATCH_START:
          case WatchEvent.WATCH_RESULT:
          case WatchEvent.WATCH_LOG:
            socket = this.getSocketByThreadId(getThreadId(worker.id, response.workerId));
            _log.debug(`session manager: watch message received from worker ${worker.id}. Type: ${response.type}`);
            if (socket === undefined) {
              _log.debug(`session manager: undefined socket. Type: ${response.type}`);
            } else {
              socket.emit(response.type, response.content);
            }
            break;
          case WatchEvent.TERMINATE_WORKER:
          default:
            break;
        }
      });
    }
  }
}

export const defaultSessionManager = new SessionManager();

defaultSessionManager.on(SESSION_EMIT_LOG, (outputType: OutputType, chunk, encoding?) => {
  const curSocket = defaultSessionManager.getActiveSocket();
  if (curSocket !== undefined) {
    curSocket.emit(BuildEvent.OUTPUT, createOutput(outputType, chunk, encoding as BufferEncoding | undefined));
  }
});
